package cn.itcast.nems.studentrecord.service.impl.studentrecord;

import cn.itcast.ic.common.exception.v2.BizExceptionProducer;
import cn.itcast.ic.common.exception.v2.bean.BizStatusEnum;
import cn.itcast.nems.common.account.AccountHolder;
import cn.itcast.nems.common.tenant.TenantInfoHolder;
import cn.itcast.nems.order.api.OrderApi;
import cn.itcast.nems.order.dto.PaymentFlowDTO;
import cn.itcast.nems.studentrecord.dto.studentrecord.*;
import cn.itcast.nems.studentrecord.service.studentrecord.StudentRecordMessageService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author liyong
 * create: 2023-11-20 16:38:54
 **/
@Slf4j
@AllArgsConstructor
@Service
public class StudentRecordMessageServiceImpl implements StudentRecordMessageService {
    private final RocketMQTemplate rocketMQTemplate;
    private final RocketMQProperties rocketMQProperties;
    private final OrderApi orderApi;
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void studentRecordCreated(StudentRecordCreateMessageDTO studentRecordCreateMessageDTO) {
        this.studentRecordMessage(studentRecordCreateMessageDTO, STUDENT_RECORD_CRETED_TAG);
    }

    @Override
    public void studentRecordPaid(StudentRecordPaidMessageDTO studentRecordPaidMessageDTO) {
        this.studentRecordMessage(studentRecordPaidMessageDTO, STUDENT_RECORD_PAID_TAG);
    }

    @Override
    public void studentRecordValid(StudentRecordValidMessageDTO studentRecordValidMessageDTO) {
        this.studentRecordMessage(studentRecordValidMessageDTO, STUDENT_RECORD_VALID_TAG);
    }

    @Override
    public void studentRecordOut(StudentRecordMessageDTO studentRecordMessageDTO) {
        this.studentRecordMessage(studentRecordMessageDTO, STUDENT_RECORD_OUT_TAG);
    }

    @Override
    public void sendOrderFlowMessages(List<PaymentFlowDTO> flows) {
        log.info("异步添加订单支付流水{}", flows);
        if (flows != null && !flows.isEmpty()) {
            flows.forEach(this::orderFlowMessage);
        }
    }

    @Override
    public void sendBalanceAmountDueChange(StudentRecordBalanceAmountDueMessageDTO balanceAmountDueMessageDTO) {
        this.studentRecordMessage(balanceAmountDueMessageDTO, STUDENT_RECORD_BALANCE_AMOUNT_DUE_TAG);
    }

    @Override
    public void receiveOrderFlowMessage(PaymentFlowDTO flow) {
        try {
            TimeUnit.SECONDS.sleep(1);
            log.info("收到订单添加流水消息'{}'", flow);
            this.orderApi.addPaymentFlow(flow);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    private void orderFlowMessage(PaymentFlowDTO flowDTO) {
        log.info("send studentRecord PaymentFlowDTO message '{}'", flowDTO);
        try {
            String tag = STUDENT_RECORD_ORDER_FLOW_TAG;
            String destination = rocketMQProperties.getConsumer().getTopic() + ":" + tag;
            String hashKey = flowDTO.getOrderId();
            final String payload = objectMapper.writeValueAsString(flowDTO);
            SendResult result = this.rocketMQTemplate.syncSendOrderly(destination, payload, hashKey);
            log.info("send studentRecord PaymentFlowDTO {} message tag '{}' result '{}'", flowDTO, tag, result);
        } catch (Exception ex) {
            log.warn(ex.getMessage(), ex);
            BizExceptionProducer.throwServerError(ex, BizStatusEnum.INTERNAL_SERVER_ERROR.toString(), "发送学籍支付消息失败'%s'", flowDTO);
        }
    }

    private void studentRecordMessage(StudentRecordMessageDTO studentRecordMessageDTO, String tag) {
        log.info("send studentRecord message '{}'", studentRecordMessageDTO);
        try {
            String destination = rocketMQProperties.getConsumer().getTopic() + ":" + tag;
            String hashKey = studentRecordMessageDTO.getStudentRecordId();
            if (studentRecordMessageDTO.getTenantId() == null || studentRecordMessageDTO.getTenantId() == 0) {
                TenantInfoHolder.getTenantInfo().ifPresent(tenantInfo -> studentRecordMessageDTO.setTenantId(tenantInfo.tenantId()));
            }
            if (studentRecordMessageDTO.getSubtenantId() == null || studentRecordMessageDTO.getSubtenantId() == 0) {
                TenantInfoHolder.getTenantInfo().ifPresent(tenantInfo -> studentRecordMessageDTO.setSubtenantId(tenantInfo.subtenantId()));
            }
            if (!StringUtils.hasText(studentRecordMessageDTO.getOperatorEmail())) {
                AccountHolder.getAccountInfo().ifPresent(a -> studentRecordMessageDTO.setOperatorEmail(a.getEmail()));
            }
            final String payload = objectMapper.writeValueAsString(studentRecordMessageDTO);
            Message<String> message = MessageBuilder
                    .withPayload(payload)
                    .setHeader(RocketMQHeaders.KEYS, hashKey)
                    .build();
            SendResult result = rocketMQTemplate.syncSendOrderly(destination, message, hashKey);
            log.info("send studentRecord '{}' message tag '{}' result '{}'", studentRecordMessageDTO.getStudentRecordId(), tag, result);
        } catch (Exception ex) {
            log.warn(ex.getMessage(), ex);
            BizExceptionProducer.throwServerError(ex, BizStatusEnum.INTERNAL_SERVER_ERROR.toString(), "发送学籍消息失败'%s'", studentRecordMessageDTO);
        }
    }
}
